feat: add class task management + csv path from config #138

rv2931 · 2024-03-24T20:12:46Z

L'idée est d'intégrer la proposition de gestion des tâches sur la base d'une classe BaseTask permettent de lancer des tâches et groupes de tâches de manière simple mais tracées avec gestion des erreurs et envoi d'email

Pour une plus grande flexibilité, le nom des fichiers CSV est migré vers un paramètre de config bloom.config, ceux-ci étant du coup surchargeable via le fichier de config .env.*

AMP_DATA_CSV_PATH=
PORT_DATA_CSV_PATH=
VESSEL_DATA_CSV_PATH=
VESSEL_POSITIONS_DATA_CSV_PATH=

Utilisation des tasks unitaires

/venv/bin/python src/tasks/facts/load_spire_data_from_json.py /path/to/file.json
/venv/bin/python src/tasks/facts/load_spire_data_from_api.py -d /path/to/dump

Utilisation des groupes de tasks (Pipeline)

/venv/bin/python src/tasks/load_dimensions.py

(venv) root@75af1fe86f04:/project# python src/tasks/load_dimensions.py
[bloom INFO @ 21:08:34] Starting task LoadDimensions
[bloom INFO @ 21:08:34] Starting task LoadAmpDataTask
[bloom INFO @ 21:08:42] Task LoadAmpDataTask finished
[bloom INFO @ 21:08:42] Starting task LoadPortDataTask
[bloom INFO @ 21:08:42] Task LoadPortDataTask finished
[bloom INFO @ 21:08:42] Starting task LoadVesselsDataTask
[bloom INFO @ 21:08:43] Task LoadVesselsDataTask finished
[bloom INFO @ 21:08:43] Task LoadDimensions finished

/venv/bin/python src/tasks/load_facts.py

(venv) root@75af1fe86f04:/project# python src/tasks/load_facts.py
[bloom INFO @ 21:10:11] Starting task LoadFacts
[bloom INFO @ 21:10:11] Starting task LoadVesselPositionsDataTask
[bloom INFO @ 21:10:12] Task LoadVesselPositionsDataTask finished
[bloom INFO @ 21:10:12] Task LoadFacts finished

/venv/bin/python src/tasks/pipeline_ports.py

(venv) root@75af1fe86f04:/project# python src/tasks/pipeline_ports.py
[bloom INFO @ 20:59:48] Starting task PipelinePorts
[bloom INFO @ 20:59:48] Starting task ComputePortGeometryBuffer
[bloom INFO @ 20:59:48] Point in time=None
[bloom INFO @ 20:59:49] 0 buffer de ports mis à jour
[bloom INFO @ 20:59:49] Task ComputePortGeometryBuffer finished
[bloom INFO @ 20:59:49] Starting task UpdateVesselDataVoyage
[bloom INFO @ 20:59:49] Point in time=2024-03-26 20:52:15.294403+00:00
[bloom INFO @ 20:59:49] 0 bateaux à traiter
[bloom INFO @ 20:59:49] 0 données SPIRE traitées
[bloom INFO @ 20:59:49] 0 données statiques mises à jour
[bloom INFO @ 20:59:49] 0 données de voyage mises à jour
[bloom INFO @ 20:59:49] Task UpdateVesselDataVoyage finished
[bloom INFO @ 20:59:49] Task PipelinePorts finished

Passage de paramètres

Les paramètres des tâches sont passés directement à la création

task=MyTask(param,mykey=value)
task.start()

Le paramètre "param" sera accessible via l'argument args de la fonction run(self,*args,**kwargs)
print(args['param'])
Le paramètre "mykey" sera accessible via l'argument kwargs de la fonction run(self,*args,**kwargs)
print(kwargs['mykey'])

Si la tâche est correctement écrite il est possible de passer les paramètres directement en ligne de commande

Les paramètres sont passés de la tâche parente à la tâche enfant en les précisant explicitement

class LoadDimensions(BaseTask):
    
    def run(self,*args,**kwargs):
        LoadAmpDataTask(*args,**kwargs).start()
        LoadPortDataTask(*args,**kwargs).start()
        LoadVesselsDataTask(*args,**kwargs).start()


if __name__ == "__main__":
    task= LoadDimensions(*list(arg for arg in sys.argv[1:] if arg.find('=') <= 0 ),
                         **dict(arg.split('=') for arg in sys.argv[1:] if arg.find('=') > 0))
    task.start()

njouanin · 2024-03-26T21:12:38Z

j'ai l'erreur suivante lorsque je lance le traitement init.py:

Traceback (most recent call last):
  File "/Users/nicolasjouanin/Dev/python/bloom/12_bloom/src/tasks/init.py", line 2, in <module>
    from tasks.data import LoadAmpDataTask, LoadPortDataTask, LoadVesselsDataTask, LoadVesselPositionsDataTask
ImportError: cannot import name 'LoadVesselPositionsDataTask' from 'tasks.data' (/Users/nicolasjouanin/Dev/python/bloom/12_bloom/src/tasks/data/__init__.py)

src/tasks/base.py

src/tasks/data/load_amp_data.py

src/tasks/data/load_port_data.py

src/tasks/data/load_vessels_data.py

src/tasks/facts/load_vessel_positions_data.py

src/tasks/load_dimensions.py

src/tasks/dimensions/load_dim_zone_amp_from_csv.py

src/tasks/load_dimensions.py

rv2931 · 2024-03-26T21:25:37Z

corrigé pour la tâche init, en fait je voulais la supprimer au profit de LoadDimensions et LoadFacts, mais au final ça peut avoir un sens que Init appelle Loaddimensions et LoadFacts donc je l'ai laissée et mise à jour

rv2931 · 2024-03-26T21:27:43Z

Oui, je voulais que tu regardes pour me dire quels traitement étaient toujours d'actualité et lesquels potentiellement pourrait grouper dans un LoadDimensions, LoadFacts, PipelineXXX et éventuel Init
Merci pour la review je regarde tout ça demain et je corrige

njouanin · 2024-03-26T21:28:48Z

Oui, je voulais que tu regardes pour me dire quels traitement étaient toujours d'actualité et lesquels potentiellement pourrait grouper dans un LoadDimensions, LoadFacts, PipelineXXX et éventuel Init

pour l'instant j'avais laissé les anciens traitements. Mais en gros tout ce qui charge directement du CSV vers des tables en passant par Dataframe est à supprimer.

rv2931 · 2024-03-26T21:29:48Z

Hésite pas à propose une autre répartition logique
Là j'ai fait

tasks/dimensions
tasks/facts
tasks/transformations
et mis les pipelines groupés à la racine de tasks mais c'est qu'une proposition

njouanin · 2024-03-26T21:29:59Z

et convert_spire_vessels_to_spire_ais_data.py ne devrait servir qu'une fois (en prod) pour convertir les données de l'ancienne table de positions vers la nouvelle.

njouanin · 2024-03-26T21:31:17Z

Hésite pas à propose une autre répartition logique Là j'ai fait
* tasks/dimensions

* tasks/facts

* tasks/transformations
  et mis les pipelines groupés à la racine de tasks mais c'est qu'une proposition

OK, je te laisse terminer, je regarderai demain 🥱... et on reparle demain à 19h en visio si tu es dispo.

rv2931 · 2024-03-28T22:50:26Z

J'ai corrigé des erreurs de syntaxe Union
Mis à jour la dock et le chargement de via docker-compose-load-data pour les tasks init & pipeline_ais_data

Par contre je termine mon chargement avec une erreur
bloom-load-data | (psycopg2.errors.ForeignKeyViolation) insert or update on table "spire_vessel_positions" violates foreign key constraint "fk_spire_vessel_positions_vessels"
bloom-load-data | DETAIL: Key (vessel_id)=(1431) is not present in table "vessels".

Y aurait un problème dans l'ordonnancement des tâches ou bien dans les fichier CSV ? @njouanin a ton avis ?

bloom-load-data  | [bloom INFO @ 22:52:48] Starting task InitTask
bloom-load-data  | [bloom INFO @ 22:52:48] Starting task LoadDimensions
bloom-load-data  | [bloom INFO @ 22:52:48] Starting task LoadDimZoneAmpFromCsv
bloom-load-data  | [bloom INFO @ 22:52:54] 720 zone AMP créés
bloom-load-data  | [bloom INFO @ 22:52:54] Task LoadDimZoneAmpFromCsv finished
bloom-load-data  | [bloom INFO @ 22:52:54] Starting task LoadDimPortFromCsv
bloom-load-data  | [bloom INFO @ 22:52:56] 5873 ports(s) créés
bloom-load-data  | [bloom INFO @ 22:52:56] Task LoadDimPortFromCsv finished
bloom-load-data  | [bloom INFO @ 22:52:56] Starting task ComputePortGeometryBuffer
bloom-load-data  | [bloom INFO @ 22:52:56] Point in time=None
bloom-load-data  | [bloom INFO @ 22:52:59] 5873 buffer de ports mis à jour
bloom-load-data  | [bloom INFO @ 22:52:59] Task ComputePortGeometryBuffer finished
bloom-load-data  | [bloom INFO @ 22:52:59] Starting task LoadDimVesselFromCsv
bloom-load-data  | [bloom INFO @ 22:53:00] 1787 bateau(x) créés
bloom-load-data  | [bloom INFO @ 22:53:00] Task LoadDimVesselFromCsv finished
bloom-load-data  | [bloom INFO @ 22:53:00] Task LoadDimensions finished
bloom-load-data  | [bloom INFO @ 22:53:00] Starting task LoadFacts
bloom-load-data  | [bloom INFO @ 22:53:00] Starting task LoadVesselPositionsDataTask
postgres_bloom   | 2024-03-28 22:53:00.462 UTC [659] ERROR:  insert or update on table "spire_vessel_positions" violates foreign key constraint "fk_spire_vessel_positions_vessels"
postgres_bloom   | 2024-03-28 22:53:00.462 UTC [659] DETAIL:  Key (vessel_id)=(1431) is not present in table "vessels".

rv2931 · 2024-03-28T23:59:53Z

je pense qu'il y a un micmac entre la table vessel et la table dim_vessel
On est d'accord:

qu'il faut utiliser vessel_subset.csv ?
que ce fichier est chargé directement dans dim_vessel ?
qu'on charge les positions avec load_spire_data_from_api ?
que load_spire_data_from_api doit chercher ses vessel dans dim_vessel et non dans la table vessel ? => si oui ce n'est pas le ca actuellement donc le chargement des positions se fini avec une erreur que le vessel 1431 n'existe pas dans vessel, effectivement il est dans dim_vessel
J'ai bon ou j'ai faux ?

njouanin · 2024-03-29T18:34:15Z

je pense qu'il y a un micmac entre la table vessel et la table dim_vessel On est d'accord:
* qu'il faut utiliser vessel_subset.csv ?

Oui

* que ce fichier est chargé directement dans dim_vessel ?

oui, via le traitement load_dim_vessel_from_csv.py

* qu'on charge les positions avec load_spire_data_from_api ?

oui, via le traitement load_spire_data_from_api.py

* que load_spire_data_from_api doit chercher ses vessel dans dim_vessel et non dans la table vessel ? => si oui ce n'est pas le ca actuellement donc le chargement des positions se fini avec une erreur que le vessel 1431 n'existe pas dans vessel, effectivement il est dans dim_vessel
  J'ai bon ou j'ai faux ?

src/tasks/dimensions/load_dim_vessel_from_csv.py

njouanin · 2024-03-30T09:02:49Z

Peut-être pour une prochaine évolution, mais pourrait-on améliorer la visibilité du logging:

rendre plus visible l'exécution des différentes étapes
marquer le début/fin de chaque étape (+ temps d'exécution)

Pourquoi pas ? pousser les log dans un channel slack....

# Conflicts: # src/bloom/tasks/clean_positions.py

Signed-off-by: herve.le-bars <[email protected]>

# Conflicts: # src/bloom/tasks/load_spire_data_from_api.py

# Conflicts: # src/bloom/tasks/update_vessel_data_voyage.py

Usage: /venv/bin/python src/task/pipeline_ports.py

# Conflicts: # src/tasks/dimensions/load_dim_vessel_from_csv.py

ré-organisation des tâches. # Conflicts: # src/tasks/dimensions/load_dim_vessel_from_csv.py

# Conflicts: # src/bloom/domain/excursion.py # src/bloom/domain/segment.py

# Conflicts: # src/tasks/dimensions/load_dim_vessel_from_csv.py

…s for csv_file_name

# Conflicts: # src/bloom/domain/segment.py

rv2931 force-pushed the task_mgmt branch 3 times, most recently from 9a6e8b0 to 8eb4337 Compare March 26, 2024 20:08

rv2931 requested a review from njouanin March 26, 2024 21:06

njouanin reviewed Mar 26, 2024

View reviewed changes

src/tasks/base.py Outdated Show resolved Hide resolved

njouanin requested changes Mar 26, 2024

View reviewed changes

njouanin approved these changes Mar 27, 2024

View reviewed changes

njouanin assigned rv2931 Mar 27, 2024

njouanin added the backend label Mar 27, 2024

rv2931 marked this pull request as ready for review March 28, 2024 20:47

rv2931 force-pushed the task_mgmt branch 4 times, most recently from 56c1164 to fc93bb3 Compare March 28, 2024 21:42

njouanin requested changes Mar 29, 2024

View reviewed changes

src/tasks/dimensions/load_dim_vessel_from_csv.py Outdated Show resolved Hide resolved

rv2931 force-pushed the task_mgmt branch from ed1294d to 6942149 Compare March 31, 2024 08:57

njouanin approved these changes Apr 1, 2024

View reviewed changes

herve.le-bars added 2 commits April 7, 2024 13:46

feat: add class task management + csv path from config

bced37a

feat: migrate clean position task

d7d0e19

# Conflicts: # src/bloom/tasks/clean_positions.py

herve.le-bars added 23 commits April 7, 2024 13:46

fix: LoadVesselPositionsDataTask

aa5570c

feat: migrate compute_port_geometry

9dd518a

Signed-off-by: herve.le-bars <[email protected]>

migrate: task ConvertSpireVesselsToSpireAisData

4c2d250

migrate: task LoadSpireDataFromApi

6fb386b

# Conflicts: # src/bloom/tasks/load_spire_data_from_api.py

migrate: task ComputePortGeometryBuffer

e65c633

migrate: task UpdateVesselDataVoyage

1a69999

# Conflicts: # src/bloom/tasks/update_vessel_data_voyage.py

feat: add pipeline PipelinePorts

0706bc1

Usage: /venv/bin/python src/task/pipeline_ports.py

fix: correction task Init (loadDimensions + LoadFacts)

0afc051

fix: stop_on_error

4c8b5ea

fix: review PR

8640322

# Conflicts: # src/tasks/dimensions/load_dim_vessel_from_csv.py

PEP8

2b5fe71

ré-organisation des tâches. # Conflicts: # src/tasks/dimensions/load_dim_vessel_from_csv.py

fix: 3.9 Union syntax

1682254

# Conflicts: # src/bloom/domain/excursion.py # src/bloom/domain/segment.py

fix: set correct registry name in docker compose file

121bad4

update: README and data-load with new tasks

fb73cde

fix: mauvais message d'erreur chargement Zone AMP

bfb4e42

fix: use vessels_subset.csv & guess CSV seperation ; or , automatically

a111e92

# Conflicts: # src/tasks/dimensions/load_dim_vessel_from_csv.py

fix: loading vessel_subset with id

351cb45

rename: extract/transform/load

5c87997

fix: LoadDimPortFromCsv/LoadDimVesselFromCsv use argument and setting…

bf7b951

…s for csv_file_name

feat: set logger format as bloom.config settings.logging_format,

edd9f80

fix/feat: load SpireDataFromCsv task

1a6ed7d

# Conflicts: # src/bloom/domain/segment.py

fix: syntax Union

f20c1ce

feat: + add batch log in columns

8087d18

rv2931 force-pushed the task_mgmt branch from c09b6ba to 8087d18 Compare April 7, 2024 11:53

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: add class task management + csv path from config #138

feat: add class task management + csv path from config #138

rv2931 commented Mar 24, 2024 •

edited

Loading

njouanin commented Mar 26, 2024

rv2931 commented Mar 26, 2024

rv2931 commented Mar 26, 2024 •

edited

Loading

njouanin commented Mar 26, 2024

rv2931 commented Mar 26, 2024

njouanin commented Mar 26, 2024

njouanin commented Mar 26, 2024

rv2931 commented Mar 28, 2024 •

edited

Loading

rv2931 commented Mar 28, 2024

njouanin commented Mar 29, 2024

njouanin commented Mar 30, 2024

feat: add class task management + csv path from config #138

Are you sure you want to change the base?

feat: add class task management + csv path from config #138

Conversation

rv2931 commented Mar 24, 2024 • edited Loading

Utilisation des tasks unitaires

Utilisation des groupes de tasks (Pipeline)

Passage de paramètres

njouanin commented Mar 26, 2024

rv2931 commented Mar 26, 2024

rv2931 commented Mar 26, 2024 • edited Loading

njouanin commented Mar 26, 2024

rv2931 commented Mar 26, 2024

njouanin commented Mar 26, 2024

njouanin commented Mar 26, 2024

rv2931 commented Mar 28, 2024 • edited Loading

rv2931 commented Mar 28, 2024

njouanin commented Mar 29, 2024

njouanin commented Mar 30, 2024

rv2931 commented Mar 24, 2024 •

edited

Loading

rv2931 commented Mar 26, 2024 •

edited

Loading

rv2931 commented Mar 28, 2024 •

edited

Loading